package com.navinfo.tripanalysis.offline.service.impl;

import com.navinfo.tripanalysis.offline.pojo.LoadDataParam;
import com.navinfo.tripanalysis.offline.service.DataBatchService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * 大数据分批处理服务
 * @author 沈东生
 */
public class DataBatchServiceImpl implements DataBatchService {
    private static final Logger LOGGER = LoggerFactory.getLogger(DataBatchServiceImpl.class);
    /**
     * 批次大小
     */
    private int batchSize;

    public int getBatchSize() {
        return batchSize;
    }

    public void setBatchSize(int batchSize) {
        this.batchSize = batchSize;
    }

    @Override
    public List<LoadDataParam> createBatch(LoadDataParam param) {
        List<LoadDataParam> list = new ArrayList<>();
        int batchSize = getBatchSize();

        //无效的批次大小，则不分批
        if (batchSize <= 0 || batchSize >= MAX) {
            LOGGER.error("batchSize:{},当前查询不分批", batchSize);
            list.add(param);
        }else{
            for (int i=0; i<MAX; i+=batchSize) {
                try {
                    //deep clone
                    LoadDataParam sub = new LoadDataParam();
                    sub.setDay(param.getDay());
                    sub.setExtendParams(new HashMap<>(param.getExtendParams()));

                    if(param.getTidList()!=null) {
                        sub.setTidList(new ArrayList<>(param.getTidList()));
                    }

                    //设置加载hashtid的分区数
                    List<Integer> subList = new ArrayList<>();
                    for (int j = i; j < Math.min(i + batchSize,MAX); j++) {
                        subList.add(j);
                    }
                    sub.setBatchList(subList);

                    list.add(sub);
                } catch (Exception e) {
                    LOGGER.error(e.getMessage(), e);
                }
            }
        }

        return list;
    }
}
